package my.thesis.bolts.persistors;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.Map;

import my.thesis.analytics.metrics.LastTimePersistMetric;

import org.joda.time.DateTime;

import redis.clients.jedis.Jedis;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public class HashStreamPersistor2 extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private Jedis jedis;
	private OutputCollector collector;
	Connection connect=null;
	Integer timeIntervalSec;
	
	transient LastTimePersistMetric _lastTimeMetric;
	private static Boolean metricsOn=false;
	
	private String database;
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		jedis=new Jedis(database,6379,60);
		
		jedis.select(8);
		this.collector=collector;
		
		if(metricsOn)
		{
			_lastTimeMetric=new LastTimePersistMetric();
			context.registerMetric("sameHashtagPersist", _lastTimeMetric, 40);
			_lastTimeMetric.setStartTime();
		}
	}
		
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		// TODO Auto-generated method stub
	}
	
	public HashStreamPersistor2(Integer ti,String connectionString)
	{
		this.timeIntervalSec=ti;
		this.database=connectionString;
	}
	
	public void execute(Tuple tuple) 
	{		
		 if(metricsOn)
		 {
			_lastTimeMetric.setTimeReceived();
		 }
		 
		 Long timeInterval=tuple.getLongByField("timeInterval");
		 
		 Long timeExactMillis=timeInterval * timeIntervalSec * 1000;
		 
		 DateTime timeExact=new DateTime(timeExactMillis);
		
		 String timeInterval_s = timeExact.toString();

		 @SuppressWarnings("unchecked")
		 ArrayList<String> sameHashesForInterval = (ArrayList<String>) tuple.getValueByField("countsForInterval");
			
		 //TEST
		 //System.out.println("Persisting same hashtags...");		

		 for(String h : sameHashesForInterval)	
		 {			 
		     jedis.lpush(timeInterval_s, h);
		 }
	}
				
	public void cleanup()
	{
	    if (jedis.isConnected()) 
	    {
		      jedis.quit();
		}	
	}

}
